package hamster.csustef.acquisition.flink.source;

import hamster.csustef.acquisition.constant.AcquisitionConstant;
import hamster.csustef.acquisition.util.TimeUtil;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Date;

public class WeatherTaskSource implements SourceFunction<String> {
    boolean running = true;
    int hour = 1000 * 60 * 60;
    int firstTerm = AcquisitionConstant.FIRST_TERM;
    int term = AcquisitionConstant.CS_WEATHER_TERM;
    int termNumber;

    public WeatherTaskSource(int runningHour) {
        this.termNumber = runningHour * hour / term;
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        long startTime = TimeUtil.sleep(firstTerm);

        while (running && startTime <= new Date().getTime() && termNumber > 0) {
            ctx.collect("ExecuteWeatherTask");
            termNumber--;
            startTime = TimeUtil.sleep(term);
        }
        ctx.collect("CLOSE");
    }

    @Override
    public void cancel() {

    }
}
